package com.zidongxiangxi.rabbit.producer.support;

import com.zidongxiangxi.rabbit.core.util.RabbitUtils;
import com.zidongxiangxi.rabbit.producer.entity.ProducerSequence;
import com.zidongxiangxi.rabbit.producer.manager.ProducerSequenceManager;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessagePostProcessor;

import java.util.Objects;

/**
 * RabbitMQ顺序消息加工
 *
 * @author chenxudong
 * @date 2020/11/07
 */
public class RabbitSequenceMessagePostProcessor implements MessagePostProcessor {
	private final Logger logger = LoggerFactory.getLogger(RabbitSequenceMessagePostProcessor.class);

	private final ProducerSequenceManager producerSequenceManager;

	public RabbitSequenceMessagePostProcessor(ProducerSequenceManager producerSequenceManager) {
		this.producerSequenceManager = producerSequenceManager;
	}

	@Override
	public Message postProcessMessage(Message message) throws AmqpException {
		if (RabbitUtils.isSequenceMessage(message)) {
			String application = RabbitUtils.getProduceApplication(message);
			String groupName = RabbitUtils.getMessageGroupName(message);
			String messageId = RabbitUtils.getMessageId(message);
			ProducerSequence sequence = producerSequenceManager.getProducer(application, messageId);
			if (Objects.nonNull(sequence)) {
				Long produceRecordId = sequence.getId();
				String previousMessageId =
						producerSequenceManager.getPreviousMessageId(application, groupName, produceRecordId);
				logger.debug("find previous message id[{}[ for application[{}] and message[{}]", previousMessageId,
						application, messageId);
				RabbitUtils.setProduceRecordId(message, produceRecordId);
				RabbitUtils.setPreviousMessageId(message, previousMessageId);
			}
		}
		return message;
	}
}
